package com.bigdata.core.example

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * checkpoint
 * 当RDD的lineage比较长，计算较为复杂时，可以使用checkpint对RDD进行持久化，checkpoint将数据直接持久化到磁盘中
 * checkpoint执行流程：
 * 1.当spark job 执行完之后会从后往前回溯，对进行checkpoint RDD进行标记
 * 2.回溯完成之后，Spark框架会启动一个job重新计算checkpointRDD的数据
 * 3.计算完成之后，将计算的结果直接持久化到指定的checkpoint目录中，切断RDD之间的依赖关系。
 * 优化：对RDD进行checkpoint之前先对RDD进行cache()下，这样第三步就不用重新从头计算当前 checkpointRDD 的数据。
 */
object Test {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("checkpoint")
    conf.setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    // 针对application，设置存储目录
    sc.setCheckpointDir("data/checkpoint")

    val lines = sc.textFile("data/words.txt")

    val lineRDD = sc.parallelize(Array("java,spark,hadoop", "java,spark", "java,spark,hadoop,spark"), 2)

    lineRDD.cache()


    val mapRDD = lineRDD.flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)

    mapRDD.collect().foreach(println)

    // mapRDD.foreach(println)

  }
}
